package com.interview.lock;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 生产者-消费者模式演示
 * 
 * 场景：200万个生产者向单个消费者发送消息
 * 
 * 核心设计思想：
 * 1. 使用ConcurrentLinkedQueue作为消息队列 - 无锁并发队列，支持高并发写入
 * 2. 生产者无需加锁，直接写入队列
 * 3. 消费者使用ReentrantLock + Condition实现等待/唤醒机制
 * 4. 批量消费提高吞吐量
 * 
 * 性能优化策略：
 * 1. 生产者端：无锁设计，使用CAS操作，避免锁竞争
 * 2. 消费者端：批量处理，减少锁获取次数
 * 3. 使用线程池管理生产者线程，避免创建过多线程
 */
public class ProducerConsumerDemo {
    
    // 生产者数量
    private static final int PRODUCER_COUNT = 2_000_000;
    // 每个生产者发送的消息数
    private static final int MESSAGES_PER_PRODUCER = 1;
    // 消费者批量处理大小
    private static final int CONSUMER_BATCH_SIZE = 1000;
    // 线程池大小（用于运行生产者）
    private static final int THREAD_POOL_SIZE = 200;
    
    public static void main(String[] args) {
        System.out.println("=== 高并发生产者-消费者模式演示 ===");
        System.out.println("生产者数量: " + PRODUCER_COUNT);
        System.out.println("每个生产者消息数: " + MESSAGES_PER_PRODUCER);
        System.out.println("总消息数: " + (PRODUCER_COUNT * MESSAGES_PER_PRODUCER));
        System.out.println("消费者批量大小: " + CONSUMER_BATCH_SIZE);
        System.out.println("线程池大小: " + THREAD_POOL_SIZE);
        System.out.println("=====================================\n");
        
        // 运行完整演示
        runFullDemo();
        
        // 可选：运行简化演示（用于快速测试）
        // runSimpleDemo();
    }
    
    /**
     * 运行完整演示（200万生产者）
     */
    private static void runFullDemo() {
        long startTime = System.currentTimeMillis();
        
        // 创建消息队列
        MessageQueue messageQueue = new MessageQueue();
        
        // 计算总消息数
        long totalMessages = (long) PRODUCER_COUNT * MESSAGES_PER_PRODUCER;
        
        // 创建并启动消费者
        Consumer consumer = new Consumer(messageQueue, CONSUMER_BATCH_SIZE, totalMessages);
        Thread consumerThread = new Thread(consumer, "Consumer-Thread");
        consumerThread.start();
        
        // 创建线程池来运行生产者
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        
        System.out.println("开始创建生产者任务...");
        
        // 创建生产者任务
        List<Producer> producers = new ArrayList<>(PRODUCER_COUNT);
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            Producer producer = new Producer(i, messageQueue, MESSAGES_PER_PRODUCER);
            producers.add(producer);
            executorService.submit(producer);
            
            // 定期打印进度
            if ((i + 1) % 100000 == 0) {
                System.out.println("已提交 " + (i + 1) + " 个生产者任务");
            }
        }
        
        System.out.println("所有生产者任务已提交\n");
        
        // 关闭线程池，不再接受新任务
        executorService.shutdown();
        
        try {
            // 等待所有生产者完成
            System.out.println("等待所有生产者完成...");
            boolean finished = executorService.awaitTermination(10, TimeUnit.MINUTES);
            
            if (finished) {
                System.out.println("所有生产者已完成\n");
            } else {
                System.out.println("生产者执行超时\n");
            }
            
            // 等待消费者处理完所有消息
            System.out.println("等待消费者处理完所有消息...");
            consumerThread.join(5 * 60 * 1000); // 最多等待5分钟
            
            // 打印最终统计信息
            messageQueue.printStats();
            
            long endTime = System.currentTimeMillis();
            long totalTime = endTime - startTime;
            
            System.out.println("\n=== 总体性能统计 ===");
            System.out.println("总耗时: " + totalTime + " 毫秒 (" + (totalTime / 1000.0) + " 秒)");
            System.out.printf("总吞吐量: %.2f 条消息/秒%n", totalMessages * 1000.0 / totalTime);
            System.out.println("==================\n");
            
        } catch (InterruptedException e) {
            System.err.println("主线程被中断: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
    
    /**
     * 运行简化演示（用于快速测试）
     * 使用较少的生产者来快速验证功能
     */
    private static void runSimpleDemo() {
        System.out.println("\n=== 简化演示（快速测试） ===\n");
        
        long startTime = System.currentTimeMillis();
        
        // 使用较少的生产者进行测试
        int simpleProducerCount = 1000;
        int simpleMessagesPerProducer = 10;
        int simpleBatchSize = 100;
        
        System.out.println("生产者数量: " + simpleProducerCount);
        System.out.println("每个生产者消息数: " + simpleMessagesPerProducer);
        System.out.println("总消息数: " + (simpleProducerCount * simpleMessagesPerProducer));
        System.out.println("消费者批量大小: " + simpleBatchSize);
        System.out.println();
        
        // 创建消息队列
        MessageQueue messageQueue = new MessageQueue();
        
        // 计算总消息数
        long totalMessages = (long) simpleProducerCount * simpleMessagesPerProducer;
        
        // 创建并启动消费者
        Consumer consumer = new Consumer(messageQueue, simpleBatchSize, totalMessages);
        Thread consumerThread = new Thread(consumer, "Consumer-Thread");
        consumerThread.start();
        
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(50);
        
        // 创建并提交生产者任务
        for (int i = 0; i < simpleProducerCount; i++) {
            Producer producer = new Producer(i, messageQueue, simpleMessagesPerProducer);
            executorService.submit(producer);
        }
        
        // 关闭线程池
        executorService.shutdown();
        
        try {
            // 等待所有生产者完成
            executorService.awaitTermination(1, TimeUnit.MINUTES);
            System.out.println("所有生产者已完成\n");
            
            // 等待消费者处理完所有消息
            consumerThread.join(1 * 60 * 1000);
            
            // 打印统计信息
            messageQueue.printStats();
            
            long endTime = System.currentTimeMillis();
            long totalTime = endTime - startTime;
            
            System.out.println("\n=== 简化演示性能统计 ===");
            System.out.println("总耗时: " + totalTime + " 毫秒");
            System.out.printf("总吞吐量: %.2f 条消息/秒%n", totalMessages * 1000.0 / totalTime);
            System.out.println("======================\n");
            
        } catch (InterruptedException e) {
            System.err.println("主线程被中断: " + e.getMessage());
            Thread.currentThread().interrupt();
        }
    }
}

